package com.easy.flux.web.rx;

import com.easy.flux.web.entity.Balance;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.function.Supplier;

/**
 * @Packge com.easy.web.rx
 * @Autor naibin.lin
 * @Description 响应式编程 demo
 * @Version 1.0
 * @Date 2022-07-29 16:22
 */
public class RxDemo {

    static Flux<Balance> list = null;
    static Flux<Balance> list2 = null;

    static {
        List<Balance> data = new ArrayList<>();
        for (long i = 1; i < 5; i++) {
            Balance balance = Balance.builder()
                    .id(i)
                    .acNo(i + "_no")
                    .acName(i + "_name")
                    .crtime(LocalDateTime.now()).build();
            data.add(balance);
        }
        list = Flux.fromIterable(data);
    }


    /**
     * 0 - 1 个元素
     */
    public static void mono() {
        Mono<String> noData = Mono.empty();
    }


    /**
     * 0 - n 个元素
     */
    public static void flux() throws InterruptedException {
        // 创建Flux或Mono，并订阅它们

        // {{{{}}}}

        list.
                map(exe -> {     // 数据处理
                    System.out.println(exe.getAcNo());
                    if (exe.getAcNo().equals("1_no")) {
                        exe.setId(998l);
                        return exe;
                    } else {
                        return exe;
                    }
                })
                // .map(Balance::getAcNo)
                //.take(2) // 只取2个
                // .delay
                .filter(e -> e.getAcNo().equals("1_no")) // 只需要 1_no , success 只有我
                .switchIfEmpty(e -> {
                    System.out.println("如果空的，执行这里...");
                })
                .log()
                .subscribe(
                        RxDemo::sucMsg,
                        RxDemo::errorMsg,
                        () -> {
                            // 结束了 调用我
                            System.out.println("complete");
                        }
                );


  /*      list.doOnNext(exe -> {
            System.out.println(exe);
        }).subscribe();*/

        //   System.out.println(list);


        //
        list.subscribe(new Subscriber<Balance>() {

            private Subscription subscription;
            private Integer count = 0;

            @Override
            public void onSubscribe(Subscription subscription) {
                this.subscription = subscription;
                //
                this.subscription.request(2); //
                System.out.println(" client request 2 onSubscribe .");

            }

            @Override
            public void onNext(Balance balance) {
                count++;
                if (count >= 2) {
                    count = 0;
                    subscription.request(2); // 请求 2 个
                    System.out.println(" client request 2 onNext .");
                }
                System.out.println(" obj ===>" + balance);
            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onComplete() {

                System.out.println(" ccc ..");
            }
        });

        // 防止 主线程过早结束
        Thread.sleep(2000l);

    }


    public static void sucMsg(Object o) {
        Balance executor = (Balance) o;
        System.out.println("=========================成功了执行我==========================" + executor);
        // executor.doSome();
        //  System.out.println("成功。。");

        //  executor.doSome();
    }

    public static void errorMsg(Throwable throwable) {
        System.err.println("=========================错误le执行我==========================" + throwable.getMessage());
    }


    public static void main(String[] args) throws Exception {
        flux();
    }
}
